package io.kiki.sba.registry.server.clientInterface.slot;

import com.google.common.collect.Sets;
import io.kiki.sba.registry.common.model.metaserver.MultiClusterSyncInfo;
import io.kiki.sba.registry.common.model.multi.cluster.RemoteSlotsStatus;
import io.kiki.sba.registry.common.model.slot.Slot;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.common.model.slot.func.SlotFunction;
import io.kiki.sba.registry.common.model.slot.func.SlotFunctionRegistry;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.shared.slot.SlotsRecorder;
import io.kiki.sba.registry.store.api.meta.MultiClusterSyncRepository;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.stream.Collectors;


public final class SlotsCacheImpl implements SlotsCache {
    private static final Logger logger = LoggerFactory.getLogger(SlotsCacheImpl.class);

    private final SlotFunction slotFunction = SlotFunctionRegistry.getSlotFunction();

    private final Lock lock = new ReentrantLock();

    private final Map<String, Slots> dataCenterToSlotsMap = new ConcurrentHashMap<>();

    @Autowired(required = false)
    private List<SlotsRecorder> slotsRecorderList;

    @Autowired
    private MultiClusterSyncRepository multiClusterSyncRepository;

    @Autowired
    private Config config;

    @PostConstruct
    public void init() {
        String dataCenter = getDataCenter();
        dataCenterToSlotsMap.put(dataCenter, Slots.INIT);
    }

    @Override
    public int slotOf(String dataInfoId) {
        return slotFunction.slotOf(dataInfoId);
    }

    @Override
    public Slot getSlot(String dataCenter, String dataInfoId) {
        int slotId = slotOf(dataInfoId);
        return getSlot(dataCenter, slotId);
    }

    @Override
    public Slot getSlot(String dataCenter, int slotId) {
        // Slots will be replace when update, not need to lock when reading
        Slots slots = this.dataCenterToSlotsMap.get(dataCenter);
        return slots == null ? null : slots.getSlot(slotId);
    }

    @Override
    public String getLeader(String dataCenter, int slotId) {
        final Slot slot = getSlot(dataCenter, slotId);
        return slot == null ? null : slot.getLeaderDataStoreNodeId();
    }

    @Override
    public long getEpoch(String dataCenter) {
        Slots slots = this.dataCenterToSlotsMap.get(dataCenter);
        return slots == null ? Slots.INIT.getEpoch() : slots.getEpoch();
    }

    @Override
    public boolean updateLocalSlots(Slots slots) {
        lock.lock();
        final long curEpoch;
        String dataCenter = getDataCenter();
        try {
            Slots exist = this.dataCenterToSlotsMap.get(dataCenter);
            if (exist == null) {
                recordSlots(slots);
                this.dataCenterToSlotsMap.put(dataCenter, slots);
                return true;
            }
            curEpoch = exist.getEpoch();
            if (curEpoch >= slots.getEpoch()) {
                logger.info("skip update, dataCenter={}, current={}, update={}", dataCenter, curEpoch, slots.getEpoch());
                return false;
            }
            recordSlots(slots);
            this.dataCenterToSlotsMap.put(dataCenter, slots);
        } finally {
            lock.unlock();
        }
        checkForSlots(dataCenter, curEpoch, slots);
        return true;
    }

    private String getDataCenter() {
        return config.getDataCenter();
    }

    @Override
    public boolean updateRemoteSlots(Map<String, RemoteSlotsStatus> remoteSlotsStatus) {
        String dataCenter = getDataCenter();
        boolean success = true;
        lock.lock();
        Set<String> difference = Sets.difference(dataCenterToSlotsMap.keySet(), remoteSlotsStatus.keySet());
        Set<String> toBeRemove = new HashSet<>(difference);
        toBeRemove.remove(dataCenter);

        try {
            for (Entry<String, RemoteSlotsStatus> entry : remoteSlotsStatus.entrySet()) {
                RemoteSlotsStatus remoteSlotsStatus1 = entry.getValue();

                String remoteDataCenter = entry.getKey();
                final long curEpoch = dataCenterToSlotsMap.computeIfAbsent(remoteDataCenter, k -> Slots.INIT).getEpoch();

                if (!remoteSlotsStatus1.isSlotsUpgrade() || remoteSlotsStatus1.getSlots() == null) {
                    logger.info("skip update, dataCenter={}, current={}, upgrade=false", remoteDataCenter, curEpoch);
                    continue;
                }

                Slots slots = remoteSlotsStatus1.getSlots();

                if (curEpoch >= slots.getEpoch()) {
                    logger.warn("skip update, current={}, update={}", curEpoch, slots.getEpoch());
                    continue;
                }
                recordSlots(slots);
                this.dataCenterToSlotsMap.put(remoteDataCenter, slots);
                logger.info("[updateRemoteSlots]dataCenter={}, prev.version={}, update.version={}, update={}", remoteDataCenter, curEpoch, slots.getEpoch(), slots);
            }
            processRemove(toBeRemove);
        } catch (Throwable throwable) {
            logger.error("update remote slot table:{} error.", remoteSlotsStatus, throwable);
            success = false;
        } finally {
            lock.unlock();
        }

        return success;
    }

    private void processRemove(Set<String> tobeRemove) {
        if (CollectionUtils.isEmpty(tobeRemove)) {
            return;
        }
        String dataCenter = getDataCenter();
        Set<MultiClusterSyncInfo> syncInfos = multiClusterSyncRepository.queryLocalMultiClusterSyncInfoSet();
        Set<String> syncing = syncInfos.stream().map(MultiClusterSyncInfo::getRemoteDataCenter).collect(Collectors.toSet());
        for (String remove : tobeRemove) {
            if (StringUtils.equals(remove, dataCenter)) {
                continue;
            }

            if (syncing.contains(remove)) {
                logger.error("dataCenter:{} remove slot table is forbidden.", remove);
                continue;
            }
            dataCenterToSlotsMap.remove(remove);
            logger.info("remove dataCenter:{} slot table success.", remove);
        }
    }

    private void recordSlots(Slots slots) {
        if (slotsRecorderList == null) {
            return;
        }
        for (SlotsRecorder recorder : slotsRecorderList) {
            if (recorder != null) {
                recorder.record(slots);
            }
        }
    }

    protected void checkForSlots(String dataCenter, long curEpoch, Slots updating) {
        for (Slot slot : updating.getSlotIdToSlotMap()) {
            if (StringUtils.isBlank(slot.getLeaderDataStoreNodeId())) {
                logger.error("[NoLeader] {},{}", dataCenter, slot);
            }
        }
        logger.info("updating slot table, dataCenter={}, expect={}, current={}, {}", dataCenter, updating.getEpoch(), curEpoch, updating);
    }

    @Override
    public Slots getLocalSlots() {
        String dataCenter = getDataCenter();
        final Slots now = dataCenterToSlotsMap.get(dataCenter);
        return new Slots(now.getEpoch(), now.getSlotIdToSlotMap());
    }

    @Override
    public Map<String, Long> getRemoteslotsEpoch() {
        if (CollectionUtils.isEmpty(dataCenterToSlotsMap)) {
            return Collections.emptyMap();
        }
        String dataCenter = getDataCenter();
        Map<String, Long> map = new HashMap<>(dataCenterToSlotsMap.size());
        for (Entry<String, Slots> entry : dataCenterToSlotsMap.entrySet()) {
            if (StringUtils.equals(entry.getKey(), dataCenter)) {
                continue;
            }
            map.put(entry.getKey(), entry.getValue().getEpoch());
        }
        return map;
    }

    @Override
    public Slots getSlots(String dataCenter) {
        final Slots slots = this.dataCenterToSlotsMap.get(dataCenter);
        return slots;
    }


}
